package com.abyss.transformation;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.util.Collector;
/**
 * @author Abyss
 * @date 2020/10/4
 * @description
 */
public class FlatMapDemo {
    public static void main(String[] args) throws Exception {
        // 1. Env
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 2. Source
        DataSource<String> source = env.fromElements(
                "张三,苹果手机,联想电脑,华为平板",
                "李四,华为手机,苹果电脑,小米平板"
        );

        // 3. 使用flatMap将数据转变为： 张三有苹果手机 张三有联想电脑... 和这样的数据
        FlatMapOperator<String, String> result = source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] strings = value.split(",");
                out.collect(strings[0] + "有" + strings[1]);
                out.collect(strings[0] + "有" + strings[2]);
                out.collect(strings[0] + "有" + strings[3]);
            }
        });

        // 4. 输出
        result.print();
    }
}
